package com.zuorenke.redis;

import com.zuorenke.help.MyCollectionUtils;
import org.apache.commons.pool2.impl.GenericObjectPool;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.Tuple;
import redis.clients.jedis.params.SetParams;

import java.util.*;

public class JedisClusterClient extends AbstractCacheClient{
    
    public static final String OK = "OK";
    
    private GenericObjectPool<JedisCluster> rcPool;
    
    public void setRcPool(GenericObjectPool<JedisCluster> rcPool) {
        this.rcPool = rcPool;
    }
    
    /**
     *
     * 描述:从连接池中获取jediscluster连接
     *
     */
    public JedisCluster getResource() {
        try {
            return rcPool.borrowObject();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     *
     * 描述:将jedisCluster连接放回连接池中
     *
     */
    public void returnResource(JedisCluster jc) {
        rcPool.returnObject(jc);
    }

    @Override
    public Boolean exists(String key) {
        JedisCluster jc = getResource();
        try {
            return jc.exists(key);
        } finally {
            returnResource(jc);
        }
    }

    @Override
    public Long expire(String key, int seconds) {
        JedisCluster jc = getResource();
        try {
            return jc.expire(key, seconds);
        } finally {
            returnResource(jc);
        }
    }

    @Override
    public Long ttl(String key) {
        JedisCluster jc = getResource();
        try {
            return jc.ttl(key);
        } finally {
            returnResource(jc);
        }
    }

    @Override
    public Set<String> keys(String pattern) {
        return MyCollectionUtils.emptySet();
    }

    @Override
    public Boolean set(String key, byte[] value) {
        String result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.set(key.getBytes(), value);
        } finally {
            returnResource(jc);
        }
        return OK.equals(result);
    }

    @Override
    public Boolean set(String key, int seconds, byte[] value) {
        String result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.setex(key.getBytes(), seconds, value);
        } finally {
            returnResource(jc);
        }
        return OK.equals(result);
    }

    @Override
    public Boolean set(String key, String value, String nxxx, long time){
        String result = null;
        JedisCluster jc = getResource();
        try {
            SetParams params = new SetParams();
            params.px(time);
            if(nxxx.equals("XX")){
                params.xx();
            }else{
                params.nx();
            }
            result = jc.set(key, value, params);
        } finally {
            returnResource(jc);
        }
        return OK.equals(result);
    }

    @Override
    public Boolean eval(String script, List<String> keys, List<String> args){
        Object result = null;
        Long releaseSuccess = 1L;
        JedisCluster jc = getResource();
        try {
            result = jc.eval(script, keys, args);
        } finally {
            returnResource(jc);
        }
        return releaseSuccess.equals(result);
    }

    @Override
    public byte[] getBytes(String key) {
        byte[] result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.get(key.getBytes());
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Boolean delete(String key) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.del(key);
        } finally {
            returnResource(jc);
        }

        return result == 1l;
    }

    @Override
    public List<Object> batchGet(final String[] keys) {
        return MyCollectionUtils.emptyList();
    }

    @Override
    public List<Object> batchDelete(final String[] keys) {
        return MyCollectionUtils.emptyList();
    }

    @Override
    public List<Object> batchSets(final List<String> keys, final List<byte[]> values) {
        return MyCollectionUtils.emptyList();
    }

    @Override
    public Long lpush(String key, byte[] value) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.lpush(key.getBytes(), value);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public List<byte[]> lrangeBytes(String key, int start, int end) {
        List<byte[]> result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.lrange(key.getBytes(), start, end);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long llen(String key) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.llen(key);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public byte[] rpopBytes(String key) {
        byte[] result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.rpop(key.getBytes());
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public byte[] lpopBytes(String key) {
        byte[] result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.lpop(key.getBytes());
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long decrBy(String key, long integer) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.decrBy(key, integer);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long incr(String key) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.incr(key);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long incrBy(String key, long integer) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.incrBy(key, integer);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Boolean hexists(String key, String field) {
        Boolean result = false;
        JedisCluster jc = getResource();
        try {
            result = jc.hexists(key, field);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Set<String> hkeys(String key) {
        Set<String> result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.hkeys(key);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long hdel(String key, String field) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.hdel(key, field);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long hsetBytes(String key, String field, byte[] value) {
        Long result = 0L;
        JedisCluster jc = getResource();
        try {
            result = jc.hset(key.getBytes(), field.getBytes(), value);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public String hmsetBytes(String key, Map<String, byte[]> map){
        return null;
    }

    @Override
    public byte[] hgetBytes(String key, String field) {
        byte[] result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.hget(key.getBytes(), field.getBytes());
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long hincrBy(String key, String field, long integer) {
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.hincrBy(key, field, integer);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public List<byte[]> hmgetBytes(String key, String... fields) {
        return MyCollectionUtils.emptyList();
    }

    @Override
    public Map<byte[], byte[]> hgetAllBytes(String key) {
        Map<byte[], byte[]> result = null;
        JedisCluster jc = getResource();

        try {
            result = jc.hgetAll(key.getBytes());
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public void publish(String channel, String message) {
        JedisCluster jc = getResource();
        try {
            jc.publish(channel, message);
        } finally {
            returnResource(jc);
        }
    }

    @Override
    public void psubscribe(JedisPubSub jedisPubSub, String[] patterns) {
        JedisCluster jc = getResource();
        try {
            jc.psubscribe(jedisPubSub, patterns);
        } finally {
            returnResource(jc);
        }
    }

    @Override
    public void flushAll() {

    }

    @Override
    public Long sadd(String key, List<Object> values){
        byte[][] byteArr = new byte[values.size()][];
        for(int i = 0; i < values.size(); i++){
            byteArr[i] = getSerializer().serialize(values.get(i));
        }
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.sadd(key.getBytes(), byteArr);
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long sadd(String key, Object value){
        Long result = null;
        JedisCluster jc = getResource();
        try {
            result = jc.sadd(key.getBytes(), getSerializer().serialize(value));
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public <T> Set<T> smembers(String key) {
        Set<T> result = new HashSet<>();
        JedisCluster jc = getResource();
        try {
            Set<byte[]> byteResult = jc.smembers(key.getBytes());
            byteResult.forEach(r ->
                result.add((T) getSerializer().deserialize(r))
            );
        } finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Boolean sismember(String key, Object value) {
        JedisCluster jc = getResource();
        Boolean result = false;
        try {
            result = jc.sismember(key.getBytes(), getSerializer().serialize(value));
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long srem(String key, Object value) {
        JedisCluster jc = getResource();
        Long result = null;
        try {
            result = jc.srem(key.getBytes(), getSerializer().serialize(value));
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long srem(String key, List<Object> values) {
        byte[][] byteArr = new byte[values.size()][];
        for(int i = 0; i < values.size(); i++){
            byteArr[i] = getSerializer().serialize(values.get(i));
        }
        JedisCluster jc = getResource();
        Long result = null;
        try {
            result = jc.srem(key.getBytes(), byteArr);
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long scard(String key){
        JedisCluster jc = getResource();
        Long result = null;
        try {
            result = jc.scard(key.getBytes());
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long zadd(String key, double score, Object member){
        byte[] byteKey = key.getBytes();
        byte[] byteMember = getSerializer().serialize(member);
        JedisCluster jc = getResource();
        Long result = null;
        try {
            result = jc.zadd(byteKey, score, byteMember);
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long zadd(String key, Map<Object, Double> scoreMembers){
        byte[] byteKey = key.getBytes();
        Map<byte[], Double> byteScoreMembers = new HashMap<>(scoreMembers.size());
        scoreMembers.forEach((k, v) ->
            byteScoreMembers.put(getSerializer().serialize(k), v)
        );
        JedisCluster jc = getResource();
        Long result = null;
        try {
            result = jc.zadd(byteKey, byteScoreMembers);
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long zrem(String key, Object value){
        byte[] byteKey = key.getBytes();
        byte[] byteValue = getSerializer().serialize(value);
        JedisCluster jc = getResource();
        Long result = null;
        try {
            result = jc.zrem(byteKey, byteValue);
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public <T> List<T> zrange(String key, long start, long end, boolean reverse){
        byte[] byteKey = key.getBytes();
        List<T> result = new ArrayList<>();
        JedisCluster jc = getResource();
        try{
            Set<byte[]> byteResult = reverse ? jc.zrevrange(byteKey, start, end) : jc.zrange(byteKey, start, end);
            byteResult.forEach(r ->
                result.add((T) getSerializer().deserialize(r))
            );
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public <T> Map<T, Double> zrangeWithScores(String key, long start, long end){
        byte[] byteKey = key.getBytes();
        Map<T, Double> result = new LinkedHashMap<>();
        JedisCluster jc = getResource();
        try {
            Set<Tuple> tuples = jc.zrangeWithScores(byteKey, start, end);
            tuples.stream().forEach(t ->
                result.put((T)getSerializer().deserialize(t.getBinaryElement()), t.getScore())
            );
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public <T> List<T> zrangeByScore(String key, double min, double max){
        byte[] byteKey = key.getBytes();
        List<T> result = new ArrayList<>();
        JedisCluster jc = getResource();
        try{
            Set<byte[]> byteResult = jc.zrangeByScore(byteKey, min, max);
            byteResult.forEach(r ->
                result.add((T) getSerializer().deserialize(r))
            );
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public <T> Map<T, Double> zrangeByScoreWithScores(String key, double min, double max){
        byte[] byteKey = key.getBytes();
        Map<T, Double> result = new LinkedHashMap<>();
        JedisCluster jc = getResource();
        try{
            Set<Tuple> tuples = jc.zrangeByScoreWithScores(byteKey, min, max);
            tuples.stream().forEach(t ->
                result.put((T)getSerializer().deserialize(t.getBinaryElement()), t.getScore())
            );
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long zcard(String key){
        byte[] byteKey = key.getBytes();
        JedisCluster jc = getResource();
        Long result = null;
        try{
            result = jc.zcard(byteKey);
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Double zscore(String key, Object member) {
        byte[] byteKey = key.getBytes();
        byte[] byteMember = getSerializer().serialize(member);
        JedisCluster jc = getResource();
        Double result = null;
        try{
            result = jc.zscore(byteKey, byteMember);
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long zcount(String key, double min, double max) {
        byte[] byteKey = key.getBytes();
        JedisCluster jc = getResource();
        Long result = null;
        try{
            result = jc.zcount(byteKey, min, max);
        }finally {
            returnResource(jc);
        }
        return result;
    }

    @Override
    public Long zrank(String key, Object member, boolean reverse) {
        byte[] byteKey = key.getBytes();
        byte[] byteMember = getSerializer().serialize(member);
        JedisCluster jc = getResource();
        Long result = null;
        try{
            result = reverse ? jc.zrevrank(byteKey, byteMember) : jc.zrank(byteKey, byteMember);
        }finally {
            returnResource(jc);
        }
        return result;
    }
}
